package com.atbeijing.D02;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import scala.Tuple2;
import scala.Tuple3;

/**
 * 求每个温度监控器的平均温度
 * reduce
 * 多个流聚合
 * 每一条元素到达以后，更新累加器的值，并将累加器的值向下游发送
 * 第一个元素到来，直接作为累加器
 */
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //流中是N个监控器的数据信息,每个监控器都会发来数据SensorReading
        DataStreamSource<SensorReading> streamSource = env.addSource(new SensorSource());

        //将对象转化为tuple,第一个值为监控器id,第二个值为温度,第三个值为计数
        SingleOutputStreamOperator<Tuple3<String, Double, Long>> mapStream = streamSource.map(new MapFunction<SensorReading, Tuple3<String, Double, Long>>() {
            @Override
            public Tuple3<String, Double, Long> map(SensorReading value) throws Exception {
                return new Tuple3<>(value.id, value.temperature, 1L);
            }
        });

        //根据监控器id分流,相当于分区
        KeyedStream<Tuple3<String, Double, Long>, String> keyedStream = mapStream.keyBy(new KeySelector<Tuple3<String, Double, Long>, String>() {
            @Override
            public String getKey(Tuple3<String, Double, Long> value) throws Exception {
                //指定分流字段
                return value._1();
            }
        });

        //根据每个分流聚合,多个聚合结果最后形成一个总流
        //第一条数据直接保存到累加器,第二条数据与累加器中的值做计算:取一个监控器id,温度相加,计数值相加.最后形成一个新的结果,传到下游
        SingleOutputStreamOperator<Tuple3<String, Double, Long>> reduceStream = keyedStream.reduce(new ReduceFunction<Tuple3<String, Double, Long>>() {
            @Override
            public Tuple3<String, Double, Long> reduce(Tuple3<String, Double, Long> value1, Tuple3<String, Double, Long> value2) throws Exception {
                return new Tuple3<>(value1._1(), value1._2() + value2._2(), value1._3() + value2._3());
            }
        });

        //转换求平均值
        SingleOutputStreamOperator<Tuple2<String, Double>> mapStream1 = reduceStream.map(new MapFunction<Tuple3<String, Double, Long>, Tuple2<String, Double>>() {
            @Override
            public Tuple2<String, Double> map(Tuple3<String, Double, Long> value) throws Exception {
                //Tuple2第一个值为监控器id,第二个值为平均值,由于是流失数据所以平均值会不断变化
                return new Tuple2<String, Double>(value._1(), value._2() / value._3());
            }
        });

        mapStream1.print();

        env.execute();
    }
}
